package topologies;

import java.io.FileInputStream;
import java.io.IOException;
import java.util.Map.Entry;
import java.util.Properties;

import spouts.CsvTweetSpout;
import spouts.StreamingTweetSpout;
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.topology.TopologyBuilder;
import bolts.CalculateUserSimilarityBolt;
import bolts.DetectTopUsersBolt;
import bolts.DetectTopWordsBolt;
import bolts.FilterNonEnglishBolt;
import bolts.OutputTopUsersBolt;
import bolts.OutputTopWordsBolt;
import bolts.OutputUserSimilarityBolt;
import bolts.TokenFrequencyByUserBolt;

public class TweetProcessingTopology {

    public static void main(String[] args) {
        final String mode = args[0];
        final int timeToRunInSeconds = Integer.valueOf(args[1]);
        final String propertiesFile = args[2];

        final Properties properties = new Properties();
        try {
            properties.load(new FileInputStream(propertiesFile));
        } catch (IOException ex) {
            System.out.println("Properties file not found");
            System.exit(-1);
        }

        Config conf = new Config();
        for (final Entry<Object, Object> property : properties.entrySet()) {
            conf.put((String) property.getKey(), property.getValue());
        }
        conf.setDebug(false);
        conf.setMaxTaskParallelism(3);

        TopologyBuilder builder = new TopologyBuilder();

        if ("csv".equals(mode)) {
            builder.setSpout("spout", new CsvTweetSpout(), 1);
        } else if ("streaming".equals(mode)) {
            builder.setSpout("spout", new StreamingTweetSpout(), 1);
        } else {
            System.out.println("Invalid mode");
            System.exit(-1);
        }

        builder.setBolt("filter", new FilterNonEnglishBolt(), 1).shuffleGrouping("spout");

        builder.setBolt("detectTopWord", new DetectTopWordsBolt("topWordsBolt", conf), 1).shuffleGrouping("filter");
        builder.setBolt("outputTopWord", new OutputTopWordsBolt(), 1).noneGrouping("detectTopWord");

        builder.setBolt("topUsers5MinBolt", new DetectTopUsersBolt("topUsers5MinBolt", conf), 1).shuffleGrouping("filter");
        builder.setBolt("outputTopUser5min", new OutputTopUsersBolt("5min"), 1).noneGrouping("topUsers5MinBolt");

        builder.setBolt("topUsers60MinBolt", new DetectTopUsersBolt("topUsers60MinBolt", conf), 1).shuffleGrouping("filter");
        builder.setBolt("outputTopUser60min", new OutputTopUsersBolt("60min"), 1).noneGrouping("topUsers60MinBolt");

        builder.setBolt("tokenFreq5MinBolt", new TokenFrequencyByUserBolt("tokenFreq5MinBolt", conf), 1).shuffleGrouping("filter");
        builder.setBolt("detectUserSimilarity5min", new CalculateUserSimilarityBolt(), 1).shuffleGrouping("tokenFreq5MinBolt");
        builder.setBolt("outputUserSimiarity5min", new OutputUserSimilarityBolt("5min"), 1).noneGrouping("detectUserSimilarity5min");

        builder.setBolt("tokenFreq20MinBolt", new TokenFrequencyByUserBolt("tokenFreq20MinBolt", conf), 1).shuffleGrouping("filter");
        builder.setBolt("detectUserSimilarity20min", new CalculateUserSimilarityBolt(), 1).shuffleGrouping("tokenFreq20MinBolt");
        builder.setBolt("outputUserSimiarity20min", new OutputUserSimilarityBolt("20min"), 1).noneGrouping("detectUserSimilarity20min");

        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("tweet-process", conf, builder.createTopology());

        try {
            Thread.sleep(1000 * timeToRunInSeconds);
        } catch (InterruptedException e) {
        }

        cluster.shutdown();
    }

}
